package beautiful.butterfly.distributed_task_processing_engine.jvm_instance_info_and_task_processor_allocation_strategy;

import beautiful.butterfly.distributed_task_processing_engine.task_manage.JvmInstanceStatusAndTaskManage;
import beautiful.butterfly.distributed_task_processing_engine.tools.Constants;
import beautiful.butterfly.distributed_task_processing_engine.tools.TimestampJsonSerializerAndJsonDeserializer;
import beautiful.butterfly.distributed_task_processing_engine.tools.Uuids;
import beautiful.butterfly.distributed_task_processing_engine.user_definition.TaskProcessorAllocationStrategyUserDefinition;
import beautiful.butterfly.distributed_task_processing_engine.zookeeper.ZooKeeperManager;
import beautiful.butterfly.distributed_task_processing_engine.zookeeper.ZooKeepers;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooKeeper;

import java.io.StringWriter;
import java.io.Writer;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;

@Slf4j
public class JvmInstanceInfoAndTaskProcessorAllocationStrategyUserDefinitionManager
{
    @Setter
    @Getter
    private ZooKeeperManager zooKeeperManager;
    @Setter
    @Getter
    private String jvmInstanceIdsRootPath;
    @Setter
    @Getter
    private String taskProcessorsRootPath;

    @Setter
    @Getter
    private Gson gson;

    /**
     * init的时候首先创建JVM根路径和任务处理器根路径
     */
    public JvmInstanceInfoAndTaskProcessorAllocationStrategyUserDefinitionManager(ZooKeeperManager zooKeeperManager) throws Exception
    {

        this.gson = new GsonBuilder().registerTypeAdapter(Timestamp.class, new TimestampJsonSerializerAndJsonDeserializer()).setDateFormat("yyyy-MM-dd HH:mm:ss").create();
        //
        this.zooKeeperManager = zooKeeperManager;
        //
        this.jvmInstanceIdsRootPath = this.zooKeeperManager.getRootPath() + Constants.$jvm_instance_ids;
        if (this.getZooKeeper().exists(this.jvmInstanceIdsRootPath, false) == null)
        {
            ZooKeepers.createPath(getZooKeeper(), this.jvmInstanceIdsRootPath, this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT);
        }
        //
        this.taskProcessorsRootPath = this.zooKeeperManager.getRootPath() + Constants.$task_processor_ids;
        if (this.getZooKeeper().exists(this.taskProcessorsRootPath, false) == null)
        {
            ZooKeepers.createPath(getZooKeeper(), this.taskProcessorsRootPath, this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT);
        }

    }


    /**
     * <pre>
     * ========================================================================================================================
     * JvmInstanceStatus
     * ========================================================================================================================
     * </pre>
     */
    /**
     * 注册自己且返回当前JVM实例需要删除的taskId(List)
     */
    public List<String> registerJvmInstanceInfoToZooKeeperAndReturnDeletedTaskIdList(JvmInstanceStatusAndTaskManage jvmInstanceStatusAndTaskManage) throws Exception
    {
        log.warn("注册" + JvmInstanceIdAndStatus.class.getSimpleName() + "信息,同时删除当前jvm需要删除的任务id(list)");

        if (jvmInstanceStatusAndTaskManage.getJvmInstanceId() == null)
        {
            log.warn("当前" + JvmInstanceIdAndStatus.class.getSimpleName() + "信息不存在");
            String uuid = Uuids.getUuid();
            String jvmInstanceIdWithOutSequentialNo =//
                    //ip不允许出现.
                    jvmInstanceStatusAndTaskManage.getIp().replace(".", "_") //
                            //hostname不允许出现-
                            + Constants.$ + jvmInstanceStatusAndTaskManage.getHostName().replace("-", "") //
                            + Constants.$ + uuid;
            String jvmInstanceIdWithOutSequentialNoPath = this.jvmInstanceIdsRootPath + Constants.slash + jvmInstanceIdWithOutSequentialNo + Constants.$;
            String jvmInstanceId = this.getZooKeeper().create(jvmInstanceIdWithOutSequentialNoPath, null, this.zooKeeperManager.getAclList(), CreateMode.EPHEMERAL_SEQUENTIAL);//瞬时节点
            jvmInstanceId = jvmInstanceId.replace(this.jvmInstanceIdsRootPath + Constants.slash, "");//去掉前面的path
            jvmInstanceStatusAndTaskManage.setJvmInstanceId(jvmInstanceId);
        } else
        {
            log.warn("当前" + JvmInstanceIdAndStatus.class.getSimpleName() + "信息存在:" + jvmInstanceStatusAndTaskManage.getJvmInstanceId());
            String path = this.jvmInstanceIdsRootPath + Constants.slash + jvmInstanceStatusAndTaskManage.getJvmInstanceId();
            if (this.getZooKeeper().exists(path, false) == null)
            {
                this.getZooKeeper().create(path, null, this.zooKeeperManager.getAclList(), CreateMode.EPHEMERAL);//瞬时节点
            }
        }
        //
        if (Constants.debug)
        {
            StringWriter stringWriter = new StringWriter();
            ZooKeepers.showInDirectoryStructureWay(this.getZooKeeper(), this.getRootPath(), stringWriter, "\r\n");
            System.out.println(stringWriter.toString());
        }
        //
        List<String> deletedTaskIdList = new ArrayList<String>();
        for (TaskProcessorAllocationStrategyUserDefinition taskProcessorAllocationStrategyUserDefinition : getTaskProcessorAllocationStrategyUserDefinitionList())//all jvm_instance_info_and_task_processor_allocation_strategy in zk
        {
            boolean allowcreateNode = false;
            //针对存活的进行管理
            if (!TaskProcessorAllocationStrategyUserDefinition.status_pause.equalsIgnoreCase(taskProcessorAllocationStrategyUserDefinition.getStatus())//
                    &&//
                    taskProcessorAllocationStrategyUserDefinition.getIpOrHostNameList() != null)//

            {
                for (String ipOrHostName : taskProcessorAllocationStrategyUserDefinition.getIpOrHostNameList())
                {
                    if (ipOrHostName.equals("127.0.0.1") || ipOrHostName.equalsIgnoreCase("localhost") ||//
                            ipOrHostName.equals(jvmInstanceStatusAndTaskManage.getIp()) || ipOrHostName.equalsIgnoreCase(jvmInstanceStatusAndTaskManage.getHostName()))
                    {//允许当前ip或者hostname在/task id/ 下创建任务处理器


                        String taskId$JvmInstanceIdPath =//
                                this.taskProcessorsRootPath +//
                                        Constants.slash + taskProcessorAllocationStrategyUserDefinition.getTaskId() + //
                                        Constants.slash + jvmInstanceStatusAndTaskManage.getJvmInstanceId();
                        allowcreateNode = true;
                        //不存在则创建taskId$JvmInstanceId节点
                        if (this.getZooKeeper().exists(taskId$JvmInstanceIdPath, false) == null)
                        {
                            this.getZooKeeper().create(taskId$JvmInstanceIdPath, null, this.zooKeeperManager.getAclList(), CreateMode.EPHEMERAL);//瞬时节点
                        }
                        break;//在这break,因为只是一个JVM
                    }
                }
            }


            //针对那些直接删除处理器分配策略,不允许直接删除,先暂停任务处理器策略,再删除.这样在暂停之后N秒后就会自动清理相关的任务处理器.(暂停后最好sleep下再删除)
            //如下两种情况需要删除:1如果是任务暂停,2如果任务是非暂停[活跃状态下]的则需要检测该任务下面的JVM实例是否应该负责处理该项任务.[这样就可以清除遗留的数据]

            if (!allowcreateNode)
            {
                String taskId$JvmInstanceIdPath =//
                        this.taskProcessorsRootPath + //
                                Constants.slash + taskProcessorAllocationStrategyUserDefinition.getTaskId() +//
                                Constants.slash + jvmInstanceStatusAndTaskManage.getJvmInstanceId();//
                if (this.getZooKeeper().exists(taskId$JvmInstanceIdPath, false) != null)
                {
                    ZooKeepers.directoryStructureDelete(this.getZooKeeper(), taskId$JvmInstanceIdPath);
                    deletedTaskIdList.add(taskProcessorAllocationStrategyUserDefinition.getTaskId());//删除在本机上面不应该允许的任务,删除方式是stop对应的处理器
                }
            }
        }
        return deletedTaskIdList;
    }


    //

    /**
     * @param active 设置当前的JVM是否有效
     */
    public void updateJvmInstanceStartStatus(String jvmInstanceId, boolean active) throws Exception
    {
        String jvmInstanceIdPath = this.jvmInstanceIdsRootPath + Constants.slash + jvmInstanceId;
        log.warn("更新jvm实例状态:" + jvmInstanceIdPath + ",状态值是:" + active);

        if (this.getZooKeeper().exists(jvmInstanceIdPath, false) == null)
        {
            System.out.println(ZooKeepers.directoryStructureGetPathList(getZooKeeper(), this.jvmInstanceIdsRootPath));
            ;
            throw new Exception("该节点不存在:" + jvmInstanceIdPath);
        }
        this.getZooKeeper().setData(jvmInstanceIdPath, Boolean.toString(active).getBytes(), -1);
    }

    /**
     * 系统所有的JvmInstanceIdAndStartStatus
     */
    public List<JvmInstanceIdAndStatus> getJvmInstanceIdAndStartStatusList() throws Exception
    {

        List<JvmInstanceIdAndStatus> jvmInstanceIdAndStatusList = new ArrayList<JvmInstanceIdAndStatus>();
        List<String> jvmInstanceIdList = this.getZooKeeper().getChildren(this.jvmInstanceIdsRootPath, false);
        Collections.sort(jvmInstanceIdList, new Comparator<String>()
        {
            public int compare(String leftString, String rightString)
            {
                return leftString.substring(leftString.lastIndexOf(Constants.$) + 1).compareTo(
                        rightString.substring(rightString.lastIndexOf(Constants.$) + 1));
            }
        });
        for (String jvmInstanceId : jvmInstanceIdList)
        {
            JvmInstanceIdAndStatus jvmInstanceIdAndStatus = new JvmInstanceIdAndStatus();
            jvmInstanceIdAndStatus.setId(jvmInstanceId);
            byte[] bytes = this.getZooKeeper().getData(this.jvmInstanceIdsRootPath + Constants.slash + jvmInstanceId, false, null);
            if (bytes == null)
            {
                jvmInstanceIdAndStatus.setActive(true);
            } else
            {
                jvmInstanceIdAndStatus.setActive(Boolean.parseBoolean(new String(bytes)));
            }
            jvmInstanceIdAndStatusList.add(jvmInstanceIdAndStatus);
        }
        return jvmInstanceIdAndStatusList;
    }

    /**
     * jvmInstanceId的JvmInstanceIdAndStartStatus
     */
    public JvmInstanceIdAndStatus getJvmInstanceIdAndStartStatus(String jvmInstanceId) throws Exception
    {
        String jvmInstanceIdPath = this.jvmInstanceIdsRootPath + Constants.slash + jvmInstanceId;
        if (this.getZooKeeper().exists(ZooKeepers.directoryStructureGetPathList(getZooKeeper(), jvmInstanceIdPath)[0], false) == null)
        {
            throw new Exception("当前节点不能存在:" + jvmInstanceIdPath);
        }
        byte[] bytes = this.getZooKeeper().getData(jvmInstanceIdPath, false, null);
        JvmInstanceIdAndStatus jvmInstanceIdAndStatus = new JvmInstanceIdAndStatus();
        jvmInstanceIdAndStatus.setId(jvmInstanceId);
        if (bytes == null)
        {
            jvmInstanceIdAndStatus.setActive(true);
        } else
        {
            jvmInstanceIdAndStatus.setActive(Boolean.parseBoolean(new String(bytes)));
        }
        return jvmInstanceIdAndStatus;
    }

    /**
     * <pre>
     * ========================================================================================================================
     * TaskProcessorAllocationStrategyUserDefinition
     *
     *
     *
     * 这里有个问题处理器分配策略暂停和恢复是在哪控制的,删除处理器分配策略,系统的处理器会怎样,这两个问题是分布式系统设计的核心。
     * ========================================================================================================================
     * </pre>
     */
    //
    public void createTaskProcessorAllocationStrategy(TaskProcessorAllocationStrategyUserDefinition taskProcessorAllocationStrategyUserDefinition) throws Exception
    {
        taskProcessorAllocationStrategyUserDefinition.setStatus(TaskProcessorAllocationStrategyUserDefinition.status_pause);//写入定义的时候暂停处理
        String taskIdPath = this.taskProcessorsRootPath + Constants.slash + taskProcessorAllocationStrategyUserDefinition.getTaskId();
        String value = this.gson.toJson(taskProcessorAllocationStrategyUserDefinition);
        if (this.getZooKeeper().exists(taskIdPath, false) != null)
        {
            throw new Exception("" + taskProcessorAllocationStrategyUserDefinition.getTaskId() + "已经存在");
        } else
        {
            this.getZooKeeper().create(taskIdPath, value.getBytes(), this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT);
        }
    }

    //
    public void pauseTaskProcessorAllocationStrategy(String taskId) throws Exception
    {
        log.warn("停止任务处理器分配策略:" + taskId);
        TaskProcessorAllocationStrategyUserDefinition taskProcessorAllocationStrategyUserDefinition = this.getTaskProcessorAllocationStrategyUserDefinition(taskId);
        taskProcessorAllocationStrategyUserDefinition.setStatus(TaskProcessorAllocationStrategyUserDefinition.status_pause);
        this.updateTaskProcessorAllocationStrategy(taskProcessorAllocationStrategyUserDefinition);
    }

    public void resumeTaskProcessorAllocationStrategy(String taskId) throws Exception
    {
        log.warn("恢复任务处理器分配策略:" + taskId);
        TaskProcessorAllocationStrategyUserDefinition taskProcessorAllocationStrategyUserDefinition = this.getTaskProcessorAllocationStrategyUserDefinition(taskId);
        taskProcessorAllocationStrategyUserDefinition.setStatus(TaskProcessorAllocationStrategyUserDefinition.status_resume);
        this.updateTaskProcessorAllocationStrategy(taskProcessorAllocationStrategyUserDefinition);
    }

    public void updateTaskProcessorAllocationStrategy(TaskProcessorAllocationStrategyUserDefinition taskProcessorAllocationStrategyUserDefinition)
            throws Exception
    {
        String taskIdPath = this.taskProcessorsRootPath + Constants.slash + taskProcessorAllocationStrategyUserDefinition.getTaskId();
        String value = this.gson.toJson(taskProcessorAllocationStrategyUserDefinition);
        if (this.getZooKeeper().exists(taskIdPath, false) == null)
        {
            this.getZooKeeper().create(taskIdPath, value.getBytes(), this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT);
        } else
        {
            this.getZooKeeper().setData(taskIdPath, value.getBytes(), -1);
        }

    }

    /**
     * 前提是必须停止调度
     */
    public void deleteTaskProcessorAllocationStrategy(String taskId) throws Exception
    {
        deleteTaskProcessorAllocationStrategy(taskId, false);
    }


    public void deleteTaskProcessorAllocationStrategy(String taskId, boolean isForce) throws Exception
    {
        String taskIdPath = this.taskProcessorsRootPath + Constants.slash + taskId;
        //isForce = true;
        if (isForce == false && this.getZooKeeper().getChildren(taskIdPath, null).size() > 0)
        {
            throw new Exception("当前" + taskId + "含有子节点");
        }
        ZooKeepers.directoryStructureDelete(this.getZooKeeper(), taskIdPath);
    }
    //

    /**
     * 返回zooKeeper中存在的处理器分配策略
     */
    public List<TaskProcessorAllocationStrategyUserDefinition> getTaskProcessorAllocationStrategyUserDefinitionList() throws Exception
    {
        log.warn("加载所有处理器分配策略");
        List<TaskProcessorAllocationStrategyUserDefinition> taskProcessorAllocationStrategyUserDefinitionList = new ArrayList<TaskProcessorAllocationStrategyUserDefinition>();
        List<String> taskIdList = this.getZooKeeper().getChildren(this.taskProcessorsRootPath, false);
        Collections.sort(taskIdList);
        for (String taskId : taskIdList)
        {
            taskProcessorAllocationStrategyUserDefinitionList.add(this.getTaskProcessorAllocationStrategyUserDefinition(taskId));
        }
        return taskProcessorAllocationStrategyUserDefinitionList;
    }

    /**
     * 经常用到的方法
     */
    public TaskProcessorAllocationStrategyUserDefinition getTaskProcessorAllocationStrategyUserDefinition(String taskId)
            throws Exception
    {
        log.warn("加载处理器分配策略:" + taskId);
        String taskIdPath = this.taskProcessorsRootPath + Constants.slash + taskId;
        if (this.getZooKeeper().exists(taskIdPath, false) == null)
        {
            return null;
        } else
        {
            String value = new String(this.getZooKeeper().getData(taskIdPath, false, null));
            TaskProcessorAllocationStrategyUserDefinition taskProcessorAllocationStrategyUserDefinition = this.gson.fromJson(value, TaskProcessorAllocationStrategyUserDefinition.class);
            return taskProcessorAllocationStrategyUserDefinition;
        }

    }

    /**
     * <pre>
     * ========================================================================================================================
     * TaskId$JvmInstanceTaskInfo
     * ========================================================================================================================
     * </pre>
     */
    public void updateTaskId$JvmInstance$ProcessorNum(String taskId, String jvmInstanceId, int processorNum) throws Exception
    {
        String taskId$JvmInstanceIdPath = this.taskProcessorsRootPath + Constants.slash + taskId + Constants.slash + jvmInstanceId;
        TaskId$JvmInstance$ProcessorNum taskId$JvmInstanceInfo = null;
        if (this.getZooKeeper().exists(taskId$JvmInstanceIdPath, false) != null)
        {
            taskId$JvmInstanceInfo = this.getTaskId$JvmInstance$ProcessorNum(taskId, jvmInstanceId);
        } else
        {
            taskId$JvmInstanceInfo = new TaskId$JvmInstance$ProcessorNum();
            taskId$JvmInstanceInfo.setTaskId(taskId);
            taskId$JvmInstanceInfo.setJvmInstanceId(jvmInstanceId);
            taskId$JvmInstanceInfo.setProcessorNum(0);//0->processorNum is the best

        }
        taskId$JvmInstanceInfo.setProcessorNum(processorNum);//设置请求数
        String value = this.gson.toJson(taskId$JvmInstanceInfo);
        this.getZooKeeper().setData(taskId$JvmInstanceIdPath, value.getBytes(), -1);
    }

    /**
     * 出现异常的时候删除/task id/jvm id
     */
    public void deleteTaskId$JvmInstanceTaskInfosWithJvmInstanceId(JvmInstanceStatusAndTaskManage jvmInstanceStatusAndTaskManage) throws Exception
    {
        log.warn("卸载处理器分配策略,当前的jvm实例信息是:" + jvmInstanceStatusAndTaskManage.getJvmInstanceId());
        for (String taskId : this.getZooKeeper().getChildren(this.taskProcessorsRootPath, false))
        {
            log.warn("卸载处理器分配策略,当前的jvm实例信息是:" + jvmInstanceStatusAndTaskManage.getJvmInstanceId() + ",当前的task id是:" + taskId);
            String taskId$JvmInstanceIdPath = this.taskProcessorsRootPath + Constants.slash + taskId + Constants.slash + jvmInstanceStatusAndTaskManage.getJvmInstanceId();
            if (this.getZooKeeper().exists(taskId$JvmInstanceIdPath, false) != null)
            {
                ZooKeepers.directoryStructureDelete(this.getZooKeeper(), taskId$JvmInstanceIdPath);
            }
        }
    }


    public List<TaskId$JvmInstance$ProcessorNum> getTaskId$JvmInstance$ProcessorNumList(String taskId) throws Exception
    {
        List<TaskId$JvmInstance$ProcessorNum> taskId$JvmInstance$ProcessorNumList = new ArrayList<TaskId$JvmInstance$ProcessorNum>();

        if (this.getZooKeeper().exists(this.taskProcessorsRootPath + Constants.slash + taskId, false) == null)
        {
            return taskId$JvmInstance$ProcessorNumList;
        }
        List<String> jvmInstanceIdList = this.getZooKeeper().getChildren(this.taskProcessorsRootPath + Constants.slash + taskId, false);
        Collections.sort(jvmInstanceIdList, new Comparator<String>()
        {
            public int compare(String leftString, String rightString)
            {
                return leftString.substring(leftString.lastIndexOf(Constants.$) + 1).compareTo(
                        rightString.substring(rightString.lastIndexOf(Constants.$) + 1));
            }
        });

        for (String jvmInstanceId : jvmInstanceIdList)
        {
            taskId$JvmInstance$ProcessorNumList.add(getTaskId$JvmInstance$ProcessorNum(taskId, jvmInstanceId));
        }
        return taskId$JvmInstance$ProcessorNumList;
    }


    public List<TaskId$JvmInstance$ProcessorNum> getTaskId$JvmInstanceTaskInfoListByJvmInstanceId(String jvmInstanceId) throws Exception
    {
        List<TaskId$JvmInstance$ProcessorNum> taskId$JvmInstance$ProcessorNumList = new ArrayList<TaskId$JvmInstance$ProcessorNum>();
        List<String> taskIdList = this.getZooKeeper().getChildren(this.taskProcessorsRootPath, false);
        Collections.sort(taskIdList);
        for (String taskId : taskIdList)
        {
            if (this.getZooKeeper().exists(this.taskProcessorsRootPath + Constants.slash + taskId + Constants.slash + jvmInstanceId, false) != null)
            {
                taskId$JvmInstance$ProcessorNumList.add(getTaskId$JvmInstance$ProcessorNum(taskId, jvmInstanceId));
            }
        }
        return taskId$JvmInstance$ProcessorNumList;
    }

    public TaskId$JvmInstance$ProcessorNum getTaskId$JvmInstance$ProcessorNum(String taskId, String jvmInstanceId) throws Exception
    {

        String taskId$JvmInstanceIdPath = this.taskProcessorsRootPath + Constants.slash + taskId + Constants.slash + jvmInstanceId;
        if (this.getZooKeeper().exists(taskId$JvmInstanceIdPath, false) == null)
        {
            return null;
        } else
        {
            byte[] bytes = this.getZooKeeper().getData(taskId$JvmInstanceIdPath, false, null);
            if (bytes != null)
            {
                String value = new String(bytes);
                TaskId$JvmInstance$ProcessorNum taskId$JvmInstance$ProcessorNum = this.gson.fromJson(value, TaskId$JvmInstance$ProcessorNum.class);
                //
                if (null == taskId$JvmInstance$ProcessorNum)
                {
                    throw new NullPointerException(TaskId$JvmInstance$ProcessorNum.class.getSimpleName());
                }
                return taskId$JvmInstance$ProcessorNum;
                //
            } else
            {
                TaskId$JvmInstance$ProcessorNum taskId$JvmInstance$ProcessorNum = new TaskId$JvmInstance$ProcessorNum();
                taskId$JvmInstance$ProcessorNum.setTaskId(taskId);
                taskId$JvmInstance$ProcessorNum.setJvmInstanceId(jvmInstanceId);
                taskId$JvmInstance$ProcessorNum.setProcessorNum(0);

                return taskId$JvmInstance$ProcessorNum;
            }
        }


    }


    public ZooKeeper getZooKeeper() throws Exception
    {
        return this.zooKeeperManager.getZooKeeper();
    }

    public String getRootPath()
    {
        return this.zooKeeperManager.getRootPath();
    }

    public void deleteTree(String path) throws Exception
    {
        ZooKeepers.directoryStructureDelete(this.getZooKeeper(), path);
    }

    public void printTreeInWay(String path, Writer writer, String lineSplit)
            throws Exception
    {
        ZooKeepers.showInDirectoryStructureWay(this.getZooKeeper(), path, writer, lineSplit);
    }


}
